package vg_connector.implementation;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.jgroups.*;
import vg_connector.interfaces.actions.*;

import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * @author Timur Zolotuhin (tzolotuhin@parallels.com)
 */
public class ChannelWorker implements Runnable {
    // Main data
    private final UUID ownerId;
    private final String clusterName;
    private final int ownerRole;

    private JChannel channel;
    private boolean close, needClose;

    private final List<ChannelListener> channelListeners;

    // Mutex
    private final Object generalMutex = new Object();

    public ChannelWorker(UUID instanceId, int ownerRole) {
        this.ownerRole = ownerRole;

        clusterName = generateClusterName(instanceId);
        ownerId = UUID.randomUUID();
        close = true;
        channelListeners = Lists.newArrayList();
    }

    public void connect() throws Exception {
        synchronized (generalMutex) {
            if (!close)
                return;

            close = false;
            needClose = false;

            channel = new JChannel("udp.xml");
            channel.connect(clusterName);

            channel.setReceiver(new ReceiverAdapter() {
                @Override
                public void receive(Message msg) {
                    try {
                        doReceiveMsg(msg);
                    } catch (Throwable ex) {
                        doNotifyErrorListeners(syncCopyChannelListeners(), ex);
                    }
                }
            });

            Thread t = new Thread(this);
            t.setDaemon(true);
            t.start();
        }
    }

    public void disconnect() {
        synchronized (generalMutex) {
            needClose = true;
        }
    }

    public boolean isConnect() {
        synchronized (generalMutex) {
            return close;
        }
    }

    public void addChannelListener(ChannelListener listener) {
        synchronized (generalMutex) {
            channelListeners.add(listener);
        }
    }

    public void removeChannelListener(ChannelListener listener) {
        synchronized (generalMutex) {
            channelListeners.add(listener);
        }
    }

    public void sendMessage(ParentMessage msg) throws Exception {
        msg.setFromId(ownerId);
        Message jGroupsMsg = new Message(null, msg);

        synchronized (generalMutex) {
            channel.send(jGroupsMsg);
        }
    }

    @Override
    public void run() {
        final Map<UUID, Integer> nodes = Maps.newHashMap();

        ChannelListener threadChannelListener = new ChannelListener() {
            @Override
            public void connect(UUID id, int type) {
                synchronized (generalMutex) {
                    nodes.put(id, 0);
                }
            }

            @Override
            public void disconnect(UUID id) {
                synchronized (generalMutex) {
                    nodes.remove(id);
                }
            }

            @Override
            public void receive(ParentMessage message) {
                if (message instanceof PingAction) {
                    PingAction pingAction = (PingAction)message;
                    PingAnswer pingAnswer = new PingAnswer(ownerRole);
                    pingAnswer.setToId(pingAction.getFromId());
                    try {
                        sendMessage(pingAnswer);
                    } catch (Throwable ex) {
                        doNotifyErrorListeners(syncCopyChannelListeners(), ex);
                    }
                }

                if (message instanceof PingAnswer) {
                    PingAnswer pingAnswer = (PingAnswer)message;
                    UUID newNodeId = null;
                    synchronized (generalMutex) {
                        if (!nodes.containsKey(pingAnswer.getFromId()))
                            newNodeId = pingAnswer.getFromId();
                        else
                            nodes.put(pingAnswer.getFromId(), 0);
                    }

                    if (newNodeId != null)
                        doNotifyConnectListeners(syncCopyChannelListeners(), newNodeId, pingAnswer.getOwnerRole());
                }
            }
        };
        addChannelListener(threadChannelListener);

        while (true) {
            synchronized (generalMutex) {
                if (needClose) {
                    channel.disconnect();
                    close = true;
                    break;
                }
            }

            // broadcast
            try {
                sendMessage(new PingAction(ownerRole));
            } catch (Throwable ex) {
                doNotifyErrorListeners(syncCopyChannelListeners(), ex);
            }

            // send ping to real node
            Map<UUID, Integer> tmpNodes;
            synchronized (generalMutex) {
                tmpNodes = Maps.newHashMap(nodes);
            }

            for (UUID id : tmpNodes.keySet()) {
                PingAction pingAction = new PingAction(ownerRole);
                pingAction.setToId(id);
                try {
                    sendMessage(pingAction);
                } catch (Throwable ex) {
                    doNotifyErrorListeners(syncCopyChannelListeners(), ex);
                }
            }

            // update number of attempts
            List<UUID> nodes4removing = Lists.newArrayList();
            synchronized (generalMutex) {
                for (UUID id : tmpNodes.keySet()) {
                    if (nodes.containsKey(id)) {
                        int attempt = nodes.get(id) + 1;
                        nodes.put(id, attempt);
                        if (attempt > 3)
                            nodes4removing.add(id);
                    }
                }
            }

            // remove old nodes
            for (UUID id : nodes4removing) {
                doNotifyDisconnectListeners(syncCopyChannelListeners(), id);
            }

            // sleep
            try {
                Thread.sleep(1000 * 3);
            } catch (Throwable ex) {
                doNotifyErrorListeners(syncCopyChannelListeners(), ex);
            }
        }
        removeChannelListener(threadChannelListener);
    }

//==============================================================================
//------------------PRIVATE METHODS---------------------------------------------
    private List<ChannelListener> syncCopyChannelListeners() {
        List<ChannelListener> copyListeners;

        synchronized (generalMutex) {
            copyListeners = Lists.newArrayList(channelListeners);
        }

        return copyListeners;
    }

    private void doNotifyConnectListeners(List<ChannelListener> connectListeners, UUID id, int type) {
        for (ChannelListener connectListener : connectListeners) {
            connectListener.connect(id, type);
        }
    }

    private void doNotifyDisconnectListeners(List<ChannelListener> connectListeners, UUID id) {
        for (ChannelListener connectListener : connectListeners) {
            connectListener.disconnect(id);
        }
    }

    private void doNotifyErrorListeners(List<ChannelListener> connectListeners, Throwable ex) {
        for (ChannelListener connectListener : connectListeners) {
            connectListener.error(ex);
        }
    }

    private void doNotifyReceiveListeners(List<ChannelListener> connectListeners, ParentMessage message) {
        for (ChannelListener connectListener : connectListeners) {
            connectListener.receive(message);
        }
    }

    private void doReceiveMsg(Message msg) {
        if (msg.getObject() instanceof ParentMessage) {
            ParentMessage parentMessage = (ParentMessage)msg.getObject();

            if (!parentMessage.getToId().equals(ownerId) && !(msg.getObject() instanceof PingAction))
                return;

            if (ownerId.equals(parentMessage.getFromId()))
                return;

            doNotifyReceiveListeners(syncCopyChannelListeners(), parentMessage);
        }
    }

    private static String generateClusterName(UUID instanceId) {
        return "cluster_id_" + instanceId.toString().replaceAll("-", "_");
    }
}
